package com.atguigu.day07;

import com.atguigu.beans.WaterSensor;
import com.atguigu.func.WaterSensorMapFunction;
import org.apache.flink.streaming.api.TimerService;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @author Felix
 * @date 2024/4/8
 * 该案例演示了定时服务
 */
public class Flink04_Timer {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<String> socketDS = env.socketTextStream("hadoop102", 8888);
        SingleOutputStreamOperator<WaterSensor> wsDS = socketDS
                .map(new WaterSensorMapFunction());
                /*.assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<WaterSensor>forMonotonousTimestamps()
                                .withTimestampAssigner(
                                        new SerializableTimestampAssigner<WaterSensor>() {
                                            @Override
                                            public long extractTimestamp(WaterSensor ws, long recordTimestamp) {
                                                return ws.getTs();
                                            }
                                        }
                                )
                );*/

        SingleOutputStreamOperator<String> processDS = wsDS.process(
                new ProcessFunction<WaterSensor, String>() {
                    @Override
                    public void processElement(WaterSensor value, ProcessFunction<WaterSensor, String>.Context ctx, Collector<String> out) throws Exception {
                        TimerService timerService = ctx.timerService();
                        long currentProcessingTime = timerService.currentProcessingTime();
                        timerService.registerProcessingTimeTimer(currentProcessingTime + 10000L);
                    }

                    @Override
                    public void onTimer(long timestamp, ProcessFunction<WaterSensor, String>.OnTimerContext ctx, Collector<String> out) throws Exception {
                        System.out.println("现在时间是" + timestamp + "定时器触发");
                    }
                }
        );

        /*//注意：如果要想使用定时服务TimeService 必须先进行keyby得到keyedStream,才能使用
        KeyedStream<WaterSensor, String> keyedDS = wsDS.keyBy(WaterSensor::getId);

        SingleOutputStreamOperator<String> processDS = keyedDS.process(
                new KeyedProcessFunction<String, WaterSensor, String>() {
                    @Override
                    public void processElement(WaterSensor ws, KeyedProcessFunction<String, WaterSensor, String>.Context ctx, Collector<String> out) throws Exception {
                        //当前元素对应的事件时间
                        Long timestamp = ctx.timestamp();
                        System.out.println("timestamp:" + timestamp);
                        //获取当前分组的key
                        String currentKey = ctx.getCurrentKey();
                        System.out.println("currentKey:" + currentKey);
                        //如果要使用侧输出流，必须要使用process方法，对流中数据进行处理，因为只有这个方法才能够将数据放到侧流中
                        //ctx.output();
                        //获取时间服务
                        TimerService timerService = ctx.timerService();

                        //获取当前水位线
                        long currentWatermark = timerService.currentWatermark();
                        System.out.println("currentWatermark:" + currentWatermark);

                        //获取当前处理时间
                        long currentProcessingTime = timerService.currentProcessingTime();
                        //注册处理时间定时器
                        timerService.registerProcessingTimeTimer(currentProcessingTime + 10000L);

                        //注册事件时间定时器
                        //timerService.registerEventTimeTimer(10);

                        //删除处理时间定时器
                        //timerService.deleteProcessingTimeTimer();
                        //删除事件时间定时器
                        //timerService.deleteEventTimeTimer();
                    }

                    @Override
                    public void onTimer(long timestamp, KeyedProcessFunction<String, WaterSensor, String>.OnTimerContext ctx, Collector<String> out) throws Exception {
                        String currentKey = ctx.getCurrentKey();
                        System.out.println("key=" + currentKey + "现在时间是" + timestamp + "定时器触发");
                    }
                }
        );*/

        env.execute();
    }
}
